package com.shujia.rec.util

import java.util.Properties

import com.shujia.rec.common.Constants
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010

object KafkaUtil {


  /**
    * 获取kafka source
    *
    * @param groupId 消费者组
    * @param topic   topic
    * @return
    */

  def getKafaSoure(groupId: String, topic: String): FlinkKafkaConsumer010[String] = {

    val kafkaProps: Properties = new Properties()
    kafkaProps.setProperty("zookeeper.connect", Constants.KAFKA_ZOOKEEPER_CONNECT)
    kafkaProps.setProperty("bootstrap.servers", Constants.KAFKA_BOOTSTRAP_SERVERS)
    kafkaProps.setProperty("group.id", groupId)
    val kafkaSource: FlinkKafkaConsumer010[String] = new FlinkKafkaConsumer010[String](topic, new SimpleStringSchema, kafkaProps)

    kafkaSource

  }

}
